/**
 * 抖音关注：程序员三丙
 * 知识星球：https://t.zsxq.com/j9b21
 */
package sanbing.example.dylike.application;

import com.fasterxml.jackson.databind.node.ObjectNode;
import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.ListenableFuture;
import jakarta.annotation.PostConstruct;
import jakarta.annotation.PreDestroy;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;
import sanbing.example.dylike.application.rpc.RpcMethod;
import sanbing.example.dylike.application.rpc.RpcRequest;
import sanbing.example.dylike.application.rpc.RpcRequestQueueService;
import sanbing.example.dylike.application.rpc.RpcResponse;
import sanbing.example.dylike.util.JacksonUtil;
import sanbing.example.dylike.util.SanbingThreadFactory;

import java.util.UUID;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;


@Component
@Slf4j
public class SeckillCmdExe {

    @Resource
    RpcRequestQueueService rpcRequestQueueService;

    private ExecutorService executor;

    @PostConstruct
    public void init() {
        executor = Executors.newSingleThreadExecutor(SanbingThreadFactory.forName("seckill-kafka-sender"));
    }

    @PreDestroy
    public void destroy() {
        if (executor != null) {
            executor.shutdownNow();
        }
    }

    public ListenableFuture<Boolean> execute(UUID skuId) {

        ObjectNode params = JacksonUtil.OBJECT_MAPPER.createObjectNode();
        params.put("skuId", skuId.toString());

        RpcRequest rpcRequest = new RpcRequest(RpcMethod.SECKILL, params);

        return Futures.transformAsync(rpcRequestQueueService.doSeckill(rpcRequest), input -> {

            RpcResponse rpcResponse = RpcResponse.from(input.getData());

            return Futures.immediateFuture(rpcResponse.isSuccess());
        }, executor);
    }
}